/*
 * Copyright 2007 Matt Jensen
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
 * the License. You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
 * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
 * specific language governing permissions and limitations under the License.
 */
package org.jtell.internal;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.jtell.*;
import org.jtell.config.Configuration;
import org.jtell.config.EventSinkMetadata;

import java.util.*;

/**
 * <p>
 * <code>EventChannelImpl</code> is the concrete implementation of the {@link EventChannel} interface.
 * </p>
 */
class EventChannelImpl implements EventChannelInternal
{
    private static Log LOG = LogFactory.getLog(EventChannelImpl.class);

    /**
     * <p>
     * The configuration. 
     * </p>
     */
    private final Configuration m_configuration;

    /**
     * <p>
     * The event metadata source. 
     * </p>
     */
    private final List<EventMetadataSource> m_eventMetadataSources;

    /**
     * <p>
     * The event notifier. 
     * </p>
     */
    private final EventNotifier m_eventNotifier;

    /**
     * <p>
     * The event sink registry. 
     * </p>
     */
    private final EventSinkRegistry m_eventSinkRegistry;

    /**
     * <p>
     * The event sink factory.
     * </p>
     */
    private final List<EventSinkSource> m_eventSinkSources;

    /**
     * <p>
     * The parent channel, if any. 
     * </p>
     */
    private final EventChannelInternal m_parentChannel;

    /**
     * <p>
     * Construct a {@link EventChannelImpl} instance.
     * </p>
     *
     * @param parentChannel the parent event channel.
     * @param configuration the configuration.
     * @param eventMetadataSources the event metadata sources.
     * @param eventSinkSources the event sink sources.
     * @param eventSinkRegistry the event sink registry.
     * @param eventNotifier the event notifier.
     */
    public EventChannelImpl(final EventChannelInternal parentChannel,
                            final Configuration configuration,
                            final List<EventMetadataSource> eventMetadataSources,
                            final EventSinkRegistry eventSinkRegistry,
                            final List<EventSinkSource> eventSinkSources,
                            final EventNotifier eventNotifier)
    {
        super();
        Guard.notNull("configuration", configuration);
        Guard.notNull("eventMetadataSources", eventMetadataSources);
        Guard.notNull("eventSinkRegistry", eventSinkRegistry);
        Guard.notNullOrEmpty("eventSinkSources", eventSinkSources);
        Guard.notNull("eventNotifier", eventNotifier);

        m_configuration = configuration;
        m_eventMetadataSources = eventMetadataSources;
        m_eventNotifier = eventNotifier;
        m_eventSinkRegistry = eventSinkRegistry;
        m_eventSinkSources = eventSinkSources;
        m_parentChannel = parentChannel;
    }

    /**
     * <p>
     * Construct a {@link EventChannelImpl} instance.
     * </p>
     *
     * @param configuration the configuration.
     * @param eventMetadataSources the event metadata sources.
     * @param eventSinkRegistry the event sink registry.
     * @param eventSinkSources the event sink sources.
     * @param eventNotifier the event notifier.
     */
    public EventChannelImpl(final Configuration configuration,
                            final List<EventMetadataSource> eventMetadataSources,
                            final EventSinkRegistry eventSinkRegistry,
                            final List<EventSinkSource> eventSinkSources,
                            final EventNotifier eventNotifier)
    {
        this(null, configuration, eventMetadataSources, eventSinkRegistry, eventSinkSources, eventNotifier);
    }

    /* inherit javadoc */
    public void addMatchingEventSinks(final Collection<EventSink> eventSinks, final EventMetadata eventMetadata)
    {
        // Add matches which were registered with this channel.
        eventSinks.addAll(m_eventSinkRegistry.findEventSinks(this, eventMetadata));

        // Add matches which were registered with the parent channel, if any.
        if (null != m_parentChannel)
        {
            m_parentChannel.addMatchingEventSinks(eventSinks, eventMetadata);
        }
    }

    /* inherit javadoc */
    public EventChannel createChildChannel()
    {
        final EventChannel result = new EventChannelImpl(this, m_configuration, m_eventMetadataSources,
                m_eventSinkRegistry, m_eventSinkSources, m_eventNotifier);
        if (LOG.isDebugEnabled())
        {
            LOG.debug(String.format("Returning new child event channel [%s] under parent [%s].", result, this));
        }
        return result;
    }

    /* inherit javadoc */
    public void notifyListeners(final Object event)
    {
        Guard.notNull("event", event);

        // Get the event metadata.
        EventMetadata eventMetadata = null;
        for (final EventMetadataSource metadataSource : m_eventMetadataSources)
        {
            eventMetadata = metadataSource.getEventMetadata(event);
            if (null != eventMetadata)
            {
                if (LOG.isDebugEnabled())
                {
                    LOG.debug(String.format("Got metadata [%s] for event [%s] from event metadata source [%s].",
                            eventMetadata, event, metadataSource));
                }
                break;
            }
        }
        if (null == eventMetadata)
        {
            throw new IllegalStateException(String.format("No event metadata source was found for event [%s].", event));
        }

        // Fill a collection with all matching event sinks.
        final List<EventSink> matchingEventSinks = new ArrayList<EventSink>();
        addMatchingEventSinks(matchingEventSinks, eventMetadata);
        if (!matchingEventSinks.isEmpty())
        {
            // Sort the event sinks based upon contribution information.
            Collections.sort(matchingEventSinks, EventSinkOrderComparator.INSTANCE);

            // Notify the listeners.
            m_eventNotifier.notifyEventSinks(matchingEventSinks, event);
        }
    }

    /* inherit javadoc */
    public void registerListener(final Object listener)
    {
        // Get the event sink metadata for this listener, if any.
        final Set<EventSinkMetadata> metadata = m_configuration.getEventSinkMetadata(listener);
        for (final EventSinkSource nextEventSinkSource : m_eventSinkSources)
        {
            final Set<EventSink> eventSinks = nextEventSinkSource.getEventSinks(listener, metadata);
            for (final EventSink nextEventSink : eventSinks)
            {
                m_eventSinkRegistry.registerEventSink(this, nextEventSink);
            }
        }
    }

    /* inherit javadoc */
    public void unregisterAllListeners()
    {
        m_eventSinkRegistry.unregisterAllEventSinks(this);
    }

    /* inherit javadoc */
    public void unregisterListener(final Object listener)
    {
        m_eventSinkRegistry.unregisterEventSinksForOwner(this, listener);
    }
}
